package main

import (
	"context"
	"fmt"
	"time"

	"github.com/segmentio/kafka-go"
)

var (
	addr  = "192.168.80.200:9092"
	topic = "test"
)

func main() {
	go Receive()
	//go Publish()

	//select{}
	time.Sleep(time.Second * 100000)
}

func Receive() {
	// make a new reader that consumes from topic-A, partition 0, at offset 42
	fmt.Println("receive connecting ...")
	r := kafka.NewReader(kafka.ReaderConfig{
		Brokers:   []string{addr},
		Topic:     topic,
		Partition: 0,
		//MinBytes:  10e3, // 10KB
		//MinBytes: 16 * 1024,        // 16KB
		MaxBytes: 32 * 1024 * 1024, // 32MB
		MaxWait:  time.Millisecond,
	})
	r.SetOffset(60)
	//ctx, _ := context.WithTimeout(context.Background(), time.Second*5)
	ctx := context.Background()
	for {
		m, err := r.ReadMessage(ctx)
		if err != nil {
			fmt.Println("readmessave error", err)
			break
		}
		//fmt.Printf("message at offset %d: %s = %s\n", m.Offset, string(m.Key), string(m.Value))
		fmt.Printf("message at offset %d: %v\n", m.Offset, m.Partition)

	}
	fmt.Println("receive close")
	r.Close()
}
func Publish() {
	fmt.Println("publish connecting ...")
	w := kafka.NewWriter(kafka.WriterConfig{
		Brokers:  []string{addr},
		Topic:    topic,
		Balancer: &kafka.LeastBytes{},
	})

	fmt.Println("publish start ...")
	for {
		message := "第1个人发送:" + time.Now().Format("2006-01-02 15:04:05")
		err := w.WriteMessages(context.Background(),
			kafka.Message{
				Key:   []byte("Key-A"),
				Value: []byte(message),
			},
		)
		if err != nil {
			print("publish err:", err)
			continue
		} else {
			fmt.Println("publish sucess")
		}

		//time.Sleep(time.Second)
	}
	//fmt.Println("public close")
	w.Close()
}
